package com.example.input

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

object FileDStream {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.ERROR)

    val conf = new SparkConf()
      .setAppName(this.getClass.getCanonicalName)
      .setMaster("local[3]")

    // 创建StreamingContext
    // StreamingContext是所有流功能函数的主要访问点，
    // 这里使用多个执行线程和10秒的批次间隔来创建本地的StreamingContext
    // 时间间隔为2秒,即2秒一个批次
    val ssc = new StreamingContext(conf, Seconds(10))

    val lines = ssc.textFileStream("G:\\data\\")
    val words = lines.flatMap(_.split("\\s+"))
    val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)
    // 打印单位时间所获得的计数值
    wordCounts.print
    ssc.start()
    // 程序会不断运行，没有会直接停止
    ssc.awaitTermination()
  }

}
